package org.rabbitmq.mq02.mqClient;

import lombok.Data;
import org.rabbitmq.mq02.common.*;
import org.rabbitmq.mq02.mqServer.core.BasicProperties;
import org.rabbitmq.mq02.mqServer.core.ExchangeType;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Data
public class Channel {
    private String channelId;
    // 当前这个 channel 属于哪个连接.
    private Connection connection;
    // 用来存储后续客户端收到的服务器的响应.
    //key:rid:请求的id, 对应的返回的结果
    private ConcurrentHashMap<String, BasicReturns> basicReturnsMap = new ConcurrentHashMap<>();
    // 如果当前 Channel 订阅了某个队列, 就需要在此处记录下对应回调是啥. 当该队列的消息返回回来的时候, 调用回调.
    // 此处约定一个 Channel 中只能有一个回调.
    private Consumer consumer = null;

    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }
    /**   type 表⽰请求响应不同的功能. 取值如下
     *  0x1  创建 channel
     * • 0x2  关闭 channel
     * • 0x3  创建 exchange
     * • 0x4  销毁 exchange
     * • 0x5  创建 queue
     * • 0x6  销毁 queue
     * • 0x7  创建 binding
     * • 0x8  销毁 binding
     * • 0x9  发送 message
     * • 0xa  订阅 message
     * • 0xb  返回 ack
     * • 0xc  服务器给客⼾端推送的消息. (被订阅的消息) 响应独有的
     */
    // 在这个方法中, 和服务器进行交互, 告知服务器, 此处客户端创建了新的 channel 了.
    public boolean createChannel() throws IOException {
        // 对于创建 Channel 操作来说, payload 就是一个 basicArgs 对象
        BasicArgs basicArgs = new BasicArgs();
        basicArgs.setChannelId(channelId);
        basicArgs.setRid(generateRid());
        byte[] payload = BinaryTool.toByte(basicArgs);

        Request request = new Request();
        request.setType(0x1);
        request.setLength(payload.length);
        request.setPayload(payload);

        // 构造出完整请求之后, 就可以发送这个请求了.
        connection.writeRequest(request);
        // 等待服务器的响应
        //服务器对根据请求处理并返回响应,对请求的处理时间不确定,
        // 该步骤可能会发生阻塞
        BasicReturns basicReturns = waitResult(basicArgs.getRid());
        return basicReturns.isOk();
    }
    // 通过UUID,生成唯一rid：请求的Id
    private String generateRid() {
        return "R-" + UUID.randomUUID().toString();
    }

    private BasicReturns waitResult(String rid) {
        BasicReturns basicReturns = null;
        while ((basicReturns = basicReturnsMap.get(rid)) == null) {
            // 如果查询结果为 null, 说明包裹还没回来.
            // 此时就需要阻塞等待.
            synchronized (this) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        // 读取成功之后, 还需要把这个消息从哈希表中删除掉.
        basicReturnsMap.remove(rid);
        return basicReturns;
    }

    public void putReturns(BasicReturns basicReturns) {
        basicReturnsMap.put(basicReturns.getRid(), basicReturns);
        synchronized (this) {
            // 当前也不知道有多少个线程在等待上述的这个响应.
            // 把所有的等待的线程都唤醒.
            notifyAll();
        }
    }

    // 关闭 channel, 给服务器发送一个 type = 0x2 的请求
    public boolean close() throws IOException {
        BasicArgs basicArgs = new BasicArgs();
        basicArgs.setRid(generateRid());
        basicArgs.setChannelId(channelId);
        byte[] payload = BinaryTool.toByte(basicArgs);

        Request request = new Request();
        request.setType(0x2);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(basicArgs.getRid());
        return basicReturns.isOk();
    }

    // 创建交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, 
               boolean durable, boolean autoDelete, Map<String, Object> Args) throws IOException {
        ExchangeDeclareArgs exchangeDeclareArgs = new ExchangeDeclareArgs();
        exchangeDeclareArgs.setRid(generateRid());
        exchangeDeclareArgs.setChannelId(channelId);
        exchangeDeclareArgs.setExchangeName(exchangeName);
        exchangeDeclareArgs.setType(exchangeType);
        exchangeDeclareArgs.setDurable(durable);
        exchangeDeclareArgs.setAutoDelete(autoDelete);
        exchangeDeclareArgs.setArgs(Args);
        byte[] payload = BinaryTool.toByte(exchangeDeclareArgs);

        Request request = new Request();
        request.setType(0x3);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(exchangeDeclareArgs.getRid());
        return basicReturns.isOk();
    }

    // 删除交换机
    public boolean exchangeDelete(String exchangeName) throws IOException {
        ExchangeDeleteArgs Args = new ExchangeDeleteArgs();
        Args.setRid(generateRid());
        Args.setChannelId(channelId);
        Args.setExchangeName(exchangeName);
        byte[] payload = BinaryTool.toByte(Args);

        Request request = new Request();
        request.setType(0x4);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(Args.getRid());
        return basicReturns.isOk();
    }

    // 创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean autoDelete,
                                Map<String, Object> Args) throws IOException {
        QueueDeclareArgs queueDeclareArgs = new QueueDeclareArgs();
        queueDeclareArgs.setRid(generateRid());
        queueDeclareArgs.setChannelId(channelId);
        queueDeclareArgs.setQueueName(queueName);
        queueDeclareArgs.setDurable(durable);
        queueDeclareArgs.setAutoDelete(autoDelete);
        queueDeclareArgs.setArgs(Args);
        byte[] payload = BinaryTool.toByte(queueDeclareArgs);

        Request request = new Request();
        request.setType(0x5);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(queueDeclareArgs.getRid());
        return basicReturns.isOk();
    }

    // 删除队列
    public boolean queueDelete(String queueName) throws IOException {
        QueueDeleteArgs Args = new QueueDeleteArgs();
        Args.setRid(generateRid());
        Args.setChannelId(channelId);
        Args.setQueueName(queueName);
        byte[] payload = BinaryTool.toByte(Args);

        Request request = new Request();
        request.setType(0x6);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(Args.getRid());
        return basicReturns.isOk();
    }

    // 创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) throws IOException {
        BindingDeclareArgs Args = new BindingDeclareArgs();
        Args.setRid(generateRid());
        Args.setChannelId(channelId);
        Args.setQueueName(queueName);
        Args.setExchangeName(exchangeName);
        Args.setBindingKey(bindingKey);
        byte[] payload = BinaryTool.toByte(Args);

        Request request = new Request();
        request.setType(0x7);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(Args.getRid());
        return basicReturns.isOk();
    }

    // 解除绑定
    public boolean queueUnbind(String queueName, String exchangeName) throws IOException {
        BindingDeleteArgs Args = new BindingDeleteArgs();
        Args.setRid(generateRid());
        Args.setChannelId(channelId);
        Args.setQueueName(queueName);
        Args.setExchangeName(exchangeName);
        byte[] payload = BinaryTool.toByte(Args);

        Request request = new Request();
        request.setType(0x8);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(Args.getRid());
        return basicReturns.isOk();
    }

    // 发送消息
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) throws IOException {
        BasicPublishArgs Args = new BasicPublishArgs();
        Args.setRid(generateRid());
        Args.setChannelId(channelId);
        Args.setExchangeName(exchangeName);
        Args.setRoutingKey(routingKey);
        Args.setBasicProperties(basicProperties);
        Args.setBody(body);
        byte[] payload = BinaryTool.toByte(Args);

        Request request = new Request();
        request.setType(0x9);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(Args.getRid());
        return basicReturns.isOk();
    }

    // 订阅消息
    public boolean basicConsume(String queueName, boolean autoAck, Consumer consumer) throws MqException, IOException {
        // 先设置回调.
        if (this.consumer != null) {
            throw new MqException("该 channel 已经设置过消费消息的回调了, 不能重复设置!");
        }
        this.consumer = consumer;

        BasicConsumerArgs Args = new BasicConsumerArgs();
        Args.setRid(generateRid());
        Args.setChannelId(channelId);
        Args.setConsumerTag(channelId);  // 此处 consumerTag 也使用 channelId 来表示了.
        Args.setQueueName(queueName);
        Args.setAutoAck(autoAck);
        byte[] payload = BinaryTool.toByte(Args);

        Request request = new Request();
        request.setType(0xa);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(Args.getRid());
        return basicReturns.isOk();
    }

    // 确认消息
    public boolean basicAck(String queueName, String messageId) throws IOException {
        BasicAckArgs Args = new BasicAckArgs();
        Args.setRid(generateRid());
        Args.setChannelId(channelId);
        Args.setQueueName(queueName);
        Args.setMessageId(messageId);
        byte[] payload = BinaryTool.toByte(Args);

        Request request = new Request();
        request.setType(0xb);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(Args.getRid());
        return basicReturns.isOk();
    }
}
